package com.atguigu.day04;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;

public class Flink05_Transfrom_ConnectKeybyDemo {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<Tuple2<Integer, String>> source1 = env.fromElements(
                Tuple2.of(1, "a1"),
                Tuple2.of(1, "a2"),
                Tuple2.of(2, "b"),
                Tuple2.of(3, "c")
        );
        DataStreamSource<Tuple3<Integer, String, Integer>> source2 = env.fromElements(
                Tuple3.of(1, "aa1", 1),
                Tuple3.of(1, "aa2", 2),
                Tuple3.of(2, "bb", 1),
                Tuple3.of(3, "cc", 1)
        );

        //关联两条流
        //连接两条流，输出能根据id匹配上的数据（类似inner join效果）
        ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> connectedStreams = source1.connect(source2);

        //将相同id的数据聚合到一起
        ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> keyBy = connectedStreams.keyBy(new KeySelector<Tuple2<Integer, String>, Integer>() {
            @Override
            public Integer getKey(Tuple2<Integer, String> value) throws Exception {
                return value.f0;
            }
        }, new KeySelector<Tuple3<Integer, String, Integer>, Integer>() {
            @Override
            public Integer getKey(Tuple3<Integer, String, Integer> value) throws Exception {
                return value.f0;
            }
        });

        //输出能根据id匹配上的数据（类似inner join效果）
        keyBy.process(new KeyedCoProcessFunction<Integer, Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>() {
            //1.定义map集合分别临时存放两条流的数据
            HashMap<Integer, List<Tuple2<Integer, String>>> tuple2Map = new HashMap<>();

            HashMap<Integer, List<Tuple3<Integer, String, Integer>>> tuple3Map = new HashMap<>();


            @Override
            public void processElement1(Tuple2<Integer, String> value, KeyedCoProcessFunction<Integer, Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
                //经过测试发现，其实不管能不能关联上 都要把自己存起来
                //存的时候要注意 是否是当前key的第一条数据，如果是第一条数据的话那么需要创建一个List集合再去存
                //如果不是第一条数据的话 那么可以直接利用之前创建好的集合add进去即可

                //判断当前数据是否是第一条这个key的数据
                if (tuple2Map.containsKey(value.f0)){
                    //不是第一条来的数据
                    //将之前的value查询出来然后将自己的数据add进去
                    List<Tuple2<Integer, String>> tuple2List = tuple2Map.get(value.f0);
                    tuple2List.add(value);
                    //再放到Map集合中等待更多数据关联自己
                    tuple2Map.put(value.f0, tuple2List);
                }else {
                    //当前这个数据是这个key第一条数据 那么需要创建(初始化)List集合
                    ArrayList<Tuple2<Integer, String>> tuple2List = new ArrayList<>();
                    tuple2List.add(value);
                    tuple2Map.put(value.f0, tuple2List);
                }

                if (tuple3Map.containsKey(value.f0)){
                    List<Tuple3<Integer, String,Integer>> values = tuple3Map.get(value.f0);
                    out.collect(values + "->" + value);
                }


               /* //1.找对方缓存是否能关联上的数据
                if (tuple3Map.containsKey(value.f0)) {
                    //有能关联上的数据
                    List<Tuple3<Integer, String,Integer>> values = tuple3Map.get(value.f0);
                    out.collect(values + "->" + value);
                    if (tuple2Map.containsKey(value.f0)){
                        List<Tuple2<Integer, String>> tuple2List = tuple2Map.get(value.f0);
                        tuple2List.add(value);
                    }else {
                        ArrayList<Tuple2<Integer, String>> tuple2s = new ArrayList<>();
                        tuple2s.add(value);
                        tuple2Map.put(value.f0, tuple2s);
                    }
                }*/
            }

            @Override
            public void processElement2(Tuple3<Integer, String, Integer> value, KeyedCoProcessFunction<Integer, Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx, Collector<String> out) throws Exception {

                if (tuple3Map.containsKey(value.f0)){
                    //不是第一条来的数据
                    //将之前的value查询出来然后将自己的数据add进去
                    List<Tuple3<Integer, String,Integer>> tuple3List = tuple3Map.get(value.f0);
                    tuple3List.add(value);
                    //再放到Map集合中等待更多数据关联自己
                    tuple3Map.put(value.f0, tuple3List);
                }else {
                    //当前这个数据是这个key第一条数据 那么需要创建(初始化)List集合
                    ArrayList<Tuple3<Integer, String,Integer>> tuple3List = new ArrayList<>();
                    tuple3List.add(value);
                    tuple3Map.put(value.f0, tuple3List);
                }

                //去对方缓存中查询是否有能关联上的数据
                if (tuple2Map.containsKey(value.f0)){
                    List<Tuple2<Integer, String>> values = tuple2Map.get(value.f0);
                    out.collect(values + "->" + value);
                }










//                //1.找对方缓存是否能关联上的数据
//                if (tuple2Map.containsKey(value.f0)) {
//                    //有能关联上的数据
//                    List<Tuple2<Integer, String>> values = tuple2Map.get(value.f0);
//                    out.collect(values + "->" + value);
//                    if (tuple3Map.containsKey(value.f0)){
//                        List<Tuple3<Integer, String, Integer>> tuple3List = tuple3Map.get(value.f0);
//                        tuple3List.add(value);
//                    }else {
//                        ArrayList<Tuple3<Integer, String, Integer>> tuple3s = new ArrayList<>();
//                        tuple3s.add(value);
//                        tuple3Map.put(value.f0, tuple3s);
//                    }
//                }
            }
        }).print();


        env.execute();

    }
}
